package com.thd.springboottest.kafka.kafkaclientedu;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.junit.Test;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

/**
 * com.thd.springboottest.kafka.kafkaclientedu.ProducerDemo
 *
 * @author: wanglei62
 * @DATE: 2022/4/8 14:21
 **/

public class ProducerDemo {

    @Test
    // 异步发送消息 -- 自动根据key分区
    public void testAsyncSend01(){
        Properties props = new Properties();
        // 服务地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 发送消息结果确认  all或-1:所有副本都收到消息才算发送成功  0:不等待确认  1:只要leader收到消息就算发送成功
        props.put(ProducerConfig.ACKS_CONFIG, "all");



        // ===== 消息发送重试配置 ======
        // 重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, 3); // 重试次数
        // 设置请求超时时间,超过这个时间会触发重试
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000);
        // 重试间隔时间  这个参数表示生产端发送请求后等待broker端响应的最长时间，默认值为30000，即30s，超时生产端可能会选择重试（如果配置了retries）
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3000); // 设置每次重试之间的等待时间为1秒(1000ms)
        // =======================

        // 设置最大阻塞时间(KafkaProducer.send()的等待时间,如果服务宕机,等待服务恢复的时间(ms), 如果超时则抛ExecutionException异常),该时间设置要大于 重试次数*重试间隔时间 (3*3000)
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);



        // ===== 批量发送配置 ======
        // 批量发送到broker的消息大小(单位时间内(ProducerConfig.LINGER_MS_CONFIG设置)积累到超过该设置的大小会一起发送给broker),单位是字节
        // 发送到缓冲区中的消息会被分为一个一个的batch，分批次的发送到broker 端，这个参数就表示batch批次大小，默认值为16384，即16KB。因此减小batch大小有利于降低消息延时，增加batch大小有利于提升吞吐量。
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 多少时间发送一次消息给broker,单位是毫秒 , 与ProducerConfig.BATCH_SIZE_CONFIG配合使用,只要积累的批量消息达到这两个条件指定的值都会发送
        // 用来控制batch最大的空闲时间，超过该时间的batch也会被发送到broker端。这实际上是一种权衡，即吞吐量与延时之间的权衡。默认值为0，表示消息需要被立即发送，无需关系batch是否被填满。
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1000);
        // =======================

        // 生产端缓冲区大小  默认值为33554432，即32M 这个参数基本可以认为是Producer程序所使用的内存大小
        // props.put("buffer.memory", 33554432);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

        // 消息key序列化器
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 消息value序列化器
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");


        // ======== ssl相关配置 ========
//        props.put("security.protocol", "SASL_SSL");
//        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "G:/Temp/client_truststore.jks"); //Save the certificate (dowload client_truststore.jks) in trust store to local server directoy (only for Java client)
//        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,"xxxxxxxx");
//        props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");
//        props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username='kaf_crm' password='xxxxxxx';"); //注意passwod结尾的分号一定不要漏
//        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");

        // =======================
        /**
         * 泛型 第一个参数是 消息key的类型  第二个参数是消息本身value的类型
         */
        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 1000; i++) {

            // ==================================== 采用同步的方式发送 ====================================

            /**
             * 采用同步的方式就是在producer.send后接收返回值即可. 返回值是Future对象,如果调用该对象的.get方法则会阻塞等待发送结果
             */

            /**
             * new ProducerRecord 三个参数为：
             * 第一个参数是 topic名称
             * 第二个参数是key (可以理解为消息的唯一标识)
             * 第三个参数是value(发送的消息内容)
             *
             * ProducerRecord泛型要与Producer泛型一致,第一个参数是 消息key的类型  第二个参数是消息本身value的类型
             */
            Future<RecordMetadata> result = producer.send(new ProducerRecord<String, String>("test", "key_"+Integer.toString(i), "value_" + Integer.toString(i)));



            RecordMetadata r = null;
            try {
                // 阻塞等待发送结果
                r = (RecordMetadata)result.get();

                // 打印出返回的结果
                System.out.println(String.format("发送成功,topic:%s,partition:%s,offset:%s",
                        r.topic(),
                        r.partition(),
                        r.offset()
                ));
            } catch (InterruptedException e) {
                System.out.println("发送失败");
                throw new RuntimeException(e);
            } catch (ExecutionException e) { // 发送超时后会抛出ExecutionException异常
                System.out.println("发送失败");
                throw new RuntimeException(e);
            }



            // ==================================== 采用异步的方式发送 ====================================
            /**
             * 采用异步的方式就是在producer.send后不接返回值即可. 返回值是Future对象,如果调用该对象的.get方法则会阻塞等待发送结果
             */
            // producer.send(new ProducerRecord<String, String>("test", "key_"+Integer.toString(i), "value_" + Integer.toString(i)));

        }
        producer.close();
    }

    @Test
    // 异步发送消息 -- 指定分区
    public void testAsyncSend02(){
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 1000; i++) {
            // 异步发送 -- 第二个参数是分区
            producer.send(new ProducerRecord<String, String>("test",Integer.valueOf(0), Integer.toString(i), Integer.toString(i)));
            System.out.println("send:" + i);
        }
        producer.close();
    }

    @Test
    // 异步发送消息 - 通过回调函数获取发送结果
    public void testAsyncSend03(){
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 1; i++) {
            // 异步发送 通过回调函数获取发送结果
            producer.send(new ProducerRecord<String, String>("test", Integer.toString(i), Integer.toString(i)),( r,e) ->{
                System.out.println(r);
                System.out.println(e);
                if(e != null){ // 如果异常参数不为空则为发送失败了
                    e.printStackTrace();
                }
            });
        }
        producer.close();
    }


    @Test
    // 同步发送消息  通过Feature.get方法阻止异步发送
    public void testSyncSend1(){
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++) {
            // 同步发送 方式1
            try{
                Future<RecordMetadata> result =  producer.send(new ProducerRecord<String, String>("test", Integer.toString(i), Integer.toString(i)));
                RecordMetadata r = result.get();
                System.out.println("发送成功:" + r);
            }catch (Exception e){
                System.out.println("发送失败:" + e.getMessage());
            }
        }
        producer.close();
    }



    @Test
    // 发送对象
    public void testSendObj(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 这里的value序列化器需要修改
        props.put("value.serializer", "org.springframework.kafka.support.serializer.JsonSerializer");

        Producer<String, JacksonBean> producer = new KafkaProducer<>(props);



        try{
            JacksonBean jb = new JacksonBean();
            jb.setAge(5);
            jb.setName("devil13th");
            jb.setBirthday(new Date());
            BigDecimal bd =new BigDecimal(5.00);
            bd.setScale(2, RoundingMode.HALF_DOWN);
            jb.setNumber(bd);

            ProducerRecord<String, JacksonBean> content = new ProducerRecord<String, JacksonBean>("testObj", "mykey", jb);

            RecordMetadata result =  producer.send(content).get();
            System.out.println("发送成功:" + result);
        }catch (Exception e){
            System.out.println("发送失败:" + e.getMessage());
        }

        producer.close();
    }

}
